from confluent_kafka import Consumer

# Kafka 配置
conf = {
    'bootstrap.servers': '192.168.66.115:9092',
    'group.id': 'Combined Lag',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': True,  # 自动提交偏移量
    'auto.commit.interval.ms': 5000,  # 每 5 秒提交一次
}

# 创建消费者实例
consumer = Consumer(conf)

# 订阅主题
topic = "ht_test"
consumer.subscribe([topic])

try:
    while True:
        msg = consumer.poll(1.0)  # 等待消息 (超时时间：1 秒)
        if msg is None:
            continue
        if msg.error():
            print(f"Consumer error: {msg.error()}")
            continue

        print(f"Received message: {msg.value().decode('utf-8')}")
except KeyboardInterrupt:
    print("Stopping consumer...")

# 关闭消费者
consumer.close()
